package streamAPI.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.testng.annotations.Test;

import java.util.concurrent.TimeUnit;

public class DemoWithSlidingWindow {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> lines = env.socketTextStream("caoyuesh1", 4444);
		SingleOutputStreamOperator<Integer> nums = lines.map(new MapFunction<String, Integer>() {
			@Override
			public Integer map(String s) throws Exception {
				return Integer.parseInt(s);
			}
		});
		// 不分组 将整体作为一个组
		nums.timeWindowAll(Time.seconds(10), Time.seconds(5)).sum(0).print();
		env.execute();

	}

	@Test()
	public void t1() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//flink,1 flink,1 hadoop,2
		DataStreamSource<String> socketTextStream = env.socketTextStream("caoyuesh1", 4444);
		SingleOutputStreamOperator<Tuple2<String, Long>> wordCountStream = socketTextStream.map(new MapFunction<String, Tuple2<String, Long>>() {
			@Override
			public Tuple2<String, Long> map(String line) throws Exception {
				String[] split = line.split(",");
				return Tuple2.of(split[0], Long.parseLong(split[1]));
			}
		});
		wordCountStream.keyBy(0).timeWindow(Time.seconds(10), Time.seconds(5)).sum(1).print();

		env.execute();
	}

	@Test(description = "基于eventTime的滑动窗口 无延迟")
	public void t2() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//设置eventTime作为时间标准
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		//1000,a,1   3000,a,1   时间戳,字母,次数
		SingleOutputStreamOperator<String> wordCount = env.socketTextStream("caoyuesh1", 4444).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
			@Override
			public long extractTimestamp(String line) {
				return Long.parseLong(line.split(",")[0]);
			}
		});
		DataStream<Tuple2<String, Long>> wordCountStream = wordCount.map(new MapFunction<String, Tuple2<String, Long>>() {
			@Override
			public Tuple2<String, Long> map(String line) throws Exception {
				String[] split = line.split(",");
				return Tuple2.of(split[1], Long.parseLong(split[2]));
			}
		});
		wordCountStream.keyBy(0).window(SlidingEventTimeWindows.of(Time.seconds(6), Time.seconds(2))).sum(1).print();
		env.execute();

	}


}
